package com.ygp.asyn.util;

import com.ygp.asyn.context.RequestAttributes;
import com.ygp.asyn.context.RequestContextHolder;
import lombok.extern.slf4j.Slf4j;

import java.time.Duration;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;
import java.util.function.Function;

@Slf4j
public class MulThreadUtil {


    public static void join(CompletableFuture<?>... futures) {
        CompletableFuture.allOf(futures).join();
    }

    public static void join(List<? extends CompletableFuture<?>> futures) {
        CompletableFuture<?>[] array;
        if (futures == null || futures.size() == 0) {
            array = new CompletableFuture<?>[]{};
        } else {
            array = futures.toArray(new CompletableFuture<?>[0]);
        }
        CompletableFuture.allOf(array).join();
    }

    public static <V> V get(CompletableFuture<V> future, long second) {
        try {
            return future.get(second, TimeUnit.SECONDS);
        } catch (Exception e) {
            log.error("get error:", e);
        }
        return null;
    }

    /**
     * 超时报错或为null
     *
     * @param future       ;
     * @param second       ;
     * @param defaultValue ;
     * @param <V>          ;
     * @return ;
     */
    public static <V> V getOrDefault(CompletableFuture<V> future, long second, V defaultValue) {
        V re = null;
        try {
            re = future.get(second, TimeUnit.SECONDS);
        } catch (Exception e) {
            log.error("get error:", e);
        }
        if (re == null) {
            return defaultValue;
        }
        return re;
    }

    public static <V> CompletableFuture<V> buildFutureWithRequestAttributes(Callable<V> callable) {
        RequestAttributes requestAttributes = RequestContextHolder.getRequestAttributes();
        return CompletableFuture.supplyAsync(() -> {
            try {
                //透传requestAttributes给子线程获取用户信息
                RequestContextHolder.setRequestAttributes(requestAttributes);
                return callable.call();
            } catch (Exception e) {
                log.error("MulThread fail ", e);
            } finally {
                RequestContextHolder.setRequestAttributes(null);
            }
            return null;
        }, CommonPoolUtil.getEXEC());
    }


    /**
     * 子线程无法获取父线程的RequestAttributes，需要透传
     *
     * @param param    ；
     * @param function ；
     * @param <R>      ；
     * @param <T>      ；
     * @return ；
     */
    public static <R, T> CompletableFuture<R> buildFutureWithRequestAttributes(T param, Function<? super T, ? extends R> function) {
        RequestAttributes requestAttributes = RequestContextHolder.getRequestAttributes();
        return CompletableFuture.supplyAsync(() -> {
            try {
                //透传requestAttributes给子线程获取用户信息
                RequestContextHolder.setRequestAttributes(requestAttributes);
                return function.apply(param);
            } catch (Exception e) {
                log.error("MulThread fail ", e);
            } finally {
                RequestContextHolder.setRequestAttributes(null);
            }
            return null;
        }, CommonPoolUtil.getEXEC());
    }


    /**
     * 无需透传的情况
     *
     * @param param    ；
     * @param function ；
     * @param <R>      ；
     * @param <T>      ；
     * @return ；
     */
    public static <R, T> CompletableFuture<R> buildFuture(T param, Function<? super T, ? extends R> function) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                return function.apply(param);
            } catch (Exception e) {
                log.error("MulThread fail ", e);
            }
            return null;
        }, CommonPoolUtil.getEXEC());
    }

    public static <R> CompletableFuture<R> buildFuture(Callable<? extends R> callable) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                return callable.call();
            } catch (Exception e) {
                log.error("MulThread fail ", e);
            }
            return null;
        }, CommonPoolUtil.getEXEC());
    }

    /**
     * 异步执行  指定超时时间
     *
     * @param param    参数
     * @param function 执行逻辑，超过后会超时
     * @param consumer 对function执行结果处理
     * @param consumer 对function执行结果处理
     * @param run      超时后执行的逻辑
     * @param <R>      ；
     * @param <T>      ；
     */
    public static <R, T> void excAsyncInTime(T param, Function<? super T, ? extends R> function, Consumer<? super R> consumer, Runnable run, Duration duration) {
        if (function == null) {
            return;
        }
        CompletableFuture<? extends R> completableFuture = buildFuture(param, function);
        CompletableFuture<? extends R> schedule = timeOutSchedule(completableFuture, duration);
        schedule.thenAccept(
                pa -> {
                    if (consumer != null) consumer.accept(pa);
                }
        )
                .exceptionally(
                        e -> {
                            if (run != null) {
                                run.run();
                            }
                            log.error("excAsyncInTime error param:{}", param, e);
                            return null;
                        }
                );
    }


    public static <R> void excAsyncInTime(Callable<? extends R> callable, Consumer<? super R> consumer, Runnable run, Duration duration) {
        if (callable == null) {
            return;
        }
        CompletableFuture<? extends R> completableFuture = buildFuture(callable);
        CompletableFuture<? extends R> schedule = timeOutSchedule(completableFuture, duration);
        schedule.thenAccept(
                pa -> {
                    if (consumer != null) consumer.accept(pa);
                }
        )
                .exceptionally(
                        e -> {
                            if (run != null) {
                                run.run();
                            }
                            log.error("excAsyncInTime error", e);
                            return null;
                        }
                );
    }


    /**
     * 获取并发结果
     *
     * @param compFuture ;
     * @param <T>        ;
     * @return ;
     */
    public static <T> T get(CompletableFuture<T> compFuture) {
        T t = null;
        try {
            t = compFuture.get();
        } catch (Exception e) {
            log.error("MulThread fail ", e);
        }
        return t;
    }

    private static <T> CompletableFuture<T> timeOutSchedule(CompletableFuture<T> taskFuture, Duration duration) {
        final CompletableFuture<T> timer = new CompletableFuture<>();
        CommonPoolUtil.getScheduler().schedule(() -> {
            TimeoutException timeoutException = new TimeoutException("time out after " + duration.getSeconds());
            return timer.completeExceptionally(timeoutException);
        }, duration.toMillis(), TimeUnit.MILLISECONDS);
        return taskFuture.applyToEither(timer, Function.identity());
    }

}


